package org.sg20.labelpropagation.lpaviewer;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.sg20.utils.LpaDotVal;
import org.sg20.utils.ParseText;



import java.io.IOException;

public class LpaViewerMapper extends Mapper<Text, Text,Text, Text> {
    private Text outKey;
    private Text outVal;
    private int propagateNum; //传播的标签数量

    @Override
    protected void setup(Mapper<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {
        super.setup(context);
        this.outKey=new Text();
        this.outVal=new Text();
        this.propagateNum=context.getConfiguration().getInt("propagate-num",1);
    }

    @Override
    protected void map(Text key, Text value, Mapper<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {
        /*
         *输入:name\t[pr1,pr2,...] [name1,weight1|name2,weight2|...]
         *输出:name label
         */
        String name=key.toString();
        String[] temp=value.toString().split(" ");
        LpaDotVal dotVal= ParseText.parseLPADotVal(temp[0]);
        int[] labels=dotVal.getLabel(propagateNum);

        if(labels==null){
            //还未有标签传播到该节点
            outVal.set("None");
        }else{
            StringBuilder sb=new StringBuilder();
            for(int label:labels){
                if(dotVal.probabilities[label]==0) break;
                sb.append(label).append(" ");
            }
            outVal.set(sb.toString());
        }

        outKey.set(name);

        context.write(outKey,outVal);
    }
}
